package com.shanks.framework.mq.job;

import com.shanks.framework.mq.common.util.DateUtil;
import com.shanks.framework.mq.common.util.MDCUtil;
import com.shanks.framework.mq.common.util.MQUtil;
import com.shanks.framework.mq.enums.MessageRecordStatusEnum;
import com.shanks.framework.mq.model.MessageRecordPO;
import com.shanks.framework.mq.queue.MessageQueueManager;
import com.shanks.framework.mq.repository.MessageRecordRepository;
import jakarta.annotation.Resource;
import net.javacrumbs.shedlock.spring.annotation.SchedulerLock;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.core.task.AsyncTaskExecutor;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.util.Date;
import java.util.List;
import java.util.concurrent.CountDownLatch;

/**
 * 消息补偿任务
 */
@Component
public class MessageCompensateJob {
    private static final Logger logger = LoggerFactory.getLogger(MessageCompensateJob.class);

    /**
     * 是否开启补偿推送
     */
    @Value("${mq-message.message-compensate-enable}")
    private boolean messageCompensateEnable;

    /**
     * 是否开启ack确认超时扫描
     */
    @Value("${mq-message.ack-checking-enable:true}")
    private boolean messageAckCheckingEnable;

    /**
     * ack超时限制重试次数
     */
    @Value("${mq-message.ack-checking-limit:2}")
    private int messageAckCheckingLimit;

    /**
     * 发送中限制重试次数
     */
    @Value("${mq-message.transmit.send-checking-limit:20160}")
    private int messageSendCheckingLimit;

    /**
     * 消息发送超时时间（秒）
     */
    @Value("${mq-message.send-timeout-seconds:30}")
    private int messageSendTimeoutSeconds;

    /**
     * 消息确认超时时间（秒）
     */
    @Value("${mq-message.ack-timeout-seconds:60}")
    private int messageAckTimeoutSeconds;


    /**
     * 分钟维度任务扫描超时时间
     */
    @Value("${job.message-repush.pending.minute.timeout.seconds}")
    private int pendingminuteTimeOutSeconds;

    /**
     * 小时维度任务扫描超时时间
     */
    @Value("${job.message-repush.pending.hour.timeout.seconds}")
    private int pendingHourTimeOutSeconds;

    /**
     * 天维度任务扫描超时时间
     */
    @Value("${job.message-repush.pending.day.timeout.seconds}")
    private int pendingDayTimeOutSeconds;

    /**
     * 消息扫描分页数
     */
    @Value("${mq-message.scan-limit}")
    private int scanLimit;

    @Value("${mq-message.ack-checking-topics:[]}")
    private List<String> ackCheckingTopics;

    @Resource
    private MessageQueueManager messageQueueManager;

    @Resource
    private MessageRecordRepository messageRecordRepository;

    /**
     * 扫描最近五分钟的消息
     */
    @Scheduled(cron = "${job.message-repush.cron}")
    @SchedulerLock(name = "scheduleRepushMessage", lockAtLeastFor = "PT15S", lockAtMostFor = "PT20S")
    public void scheduleMinuteRepushMessage() {
        executeTask("秒维度", -5, pendingminuteTimeOutSeconds, false);  // 新增一个参数来控制是否处理 PENDING 消息
    }

    /**
     * 处理过去 1 小时内的消息
     */
    @SchedulerLock(name = "scheduleRepushhHourMessage", lockAtLeastFor = "PT55S", lockAtMostFor = "PT60S")
    @Scheduled(cron = "${job.message-repush.pending.hour.cron}")
    public void scheduleRepublishHourMessage() {
        executeTask("小时维度", -60, pendingHourTimeOutSeconds, true);  // 保持处理 PENDING 消息
    }

    /**
     * 处理过去 2 天内的消息
     */
    @SchedulerLock(name = "scheduleRepushDayMessage", lockAtLeastFor = "PT55S", lockAtMostFor = "PT60S")
    @Scheduled(cron = "${job.message-repush.pending.day.cron}")
    public void scheduleRepublishDayMessage() {
        executeTask("天维度", -2880, pendingDayTimeOutSeconds, true);  // 保持处理 PENDING 消息
    }

    private void executeTask(String dimension, int timeOffsetMinutes, int timeoutSeconds, boolean handlePending) {
        if (!messageCompensateEnable) {
            logger.info("消息检查重发任务被禁用({})", dimension);
            return;
        }

        logger.info("{}任务开始执行", dimension);
        try {
            Date now = new Date();
            Date startTime = DateUtil.addDateMinute(now, timeOffsetMinutes);
            Date endTime = now;
            int timeOutLimit = timeoutSeconds * 1000;

            if (handlePending) {
                // 如果需要处理 PENDING 消息
                processMessages(startTime, endTime, timeOutLimit, MessageRecordStatusEnum.PENDING);
            }
            // 处理 SENDING 和 SENT 消息
            processMessages(startTime, endTime, timeOutLimit, MessageRecordStatusEnum.SENDING);
            if (messageAckCheckingEnable) {
                processMessages(startTime, endTime, timeOutLimit, MessageRecordStatusEnum.SENT);
            }
        } catch (Exception e) {
            logger.error("{}任务执行异常", dimension, e);
        } finally {
            logger.info("{}任务结束", dimension);
        }
    }

    private void processMessages(Date startTime, Date endTime, int timeoutMillis, MessageRecordStatusEnum status) {
        long start = System.currentTimeMillis();
        int pageIndex = 1;
        List<MessageRecordPO> recordList = fetchMessages(startTime, endTime, status, pageIndex);

        while (!CollectionUtils.isEmpty(recordList)) {
            logger.info("检测到{}状态的消息数量：{}", status, recordList.size());

            for (MessageRecordPO record : recordList) {
                handleMessage(record, status);
            }

            if (System.currentTimeMillis() - start > timeoutMillis) {
                logger.warn("处理{}状态消息超时，中断本次处理", status);
                break;
            }

            pageIndex++;
            recordList = fetchMessages(startTime, endTime, status, pageIndex);
        }
    }

    private List<MessageRecordPO> fetchMessages(Date startTime, Date endTime, MessageRecordStatusEnum status, int pageIndex) {
        return switch (status) {
            case PENDING ->
                    messageRecordRepository.selectPendingMessageList(startTime, endTime, scanLimit, pageIndex, MQUtil.mQProducerKeys);
            case SENDING ->
                    messageRecordRepository.selectSendTimeoutMessageList(startTime, endTime, scanLimit, pageIndex, MQUtil.mQProducerKeys);
            case SENT ->
                    messageRecordRepository.selectAckTimeoutMessageList(startTime, endTime, scanLimit, pageIndex, MQUtil.mQProducerKeys, ackCheckingTopics);
            default -> throw new IllegalArgumentException("Unsupported status: " + status);
        };
    }

    private void handleMessage(MessageRecordPO record, MessageRecordStatusEnum status) {
        try {
            if (record.getRetryCount().shortValue() >= messageAckCheckingLimit) {
                if (messageRecordRepository.updateMessageStatus(record.getMessageId(), MessageRecordStatusEnum.SENT, MessageRecordStatusEnum.ACK_TIMEOUT))
                    logger.error("消息发送次数超过Ack超时最大尝试次数，修改状态为:{}, messageId: {}, messageData: {}", MessageRecordStatusEnum.ACK_TIMEOUT, record
                            .getMessageId(), record.getMessageJson());
            }
            if (status == MessageRecordStatusEnum.PENDING) {
                messageQueueManager.offerMessageIfAbsent(record, MessageRecordStatusEnum.PENDING);
            } else if (status == MessageRecordStatusEnum.SENDING && record.getRetryCount() < messageSendCheckingLimit) {
                messageQueueManager.offerMessageIfAbsent(record, MessageRecordStatusEnum.SENDING);
            } else if (status == MessageRecordStatusEnum.SENT && record.getRetryCount() < messageAckCheckingLimit) {
                messageQueueManager.offerMessageIfAbsent(record, MessageRecordStatusEnum.SENT);
            } else {
                logger.warn("消息重试次数超限，ID: {}", record.getMessageId());
            }
        } catch (Exception e) {
            logger.error("处理消息异常，ID: {}", record.getMessageId(), e);
        }
    }

}
